package com.cloudfun.msglisten.ws.haier;


import com.alibaba.fastjson.JSONObject;
import com.cloudfun.msglisten.common.model.dto.dingtalk.TextContent;
import com.cloudfun.msglisten.common.model.haier.HaierBaseReqDTO;
import com.cloudfun.msglisten.common.model.haier.HaierWsSubReqDTO;
import com.cloudfun.msglisten.service.DingTalkService;
import com.cloudfun.msglisten.util.haier.UhomeUtils;
import com.cloudfun.msglisten.ws.haier.config.HaierWsProperties;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import java.net.URI;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Slf4j
@Component
@ClientEndpoint(encoders = {TextEncoder.class})
public class HaierWsClient implements InitializingBean, DisposableBean {

    // 创建容器对象（建立连接时要用）
    private WebSocketContainer container;

    // 一个连接对应一个 session
    private Session session;

    @Autowired
    private HaierWsProperties haierWsProperties;

    @Autowired
    private HaierWsClientHeartBeatService haierWsClientHeartBeatService;

    @Autowired
    private HaierWsMsgDispatcher haierWsMsgDispatcher;

    @Autowired
    private DingTalkService dingTalkService;

    private static final int connFailRetryTimes = 8;
    private static int connFailRetriedTimes = 0;

    @SneakyThrows
    public void updateConnectResult(boolean connResult) {
        if (!connResult) {
            log.error("海尔websocket认证失败");
            // 重试
            if (connFailRetriedTimes <= connFailRetryTimes) {
                TimeUnit.SECONDS.sleep(20);
                log.error("海尔websocket认证失败，尝试重连：{}", connFailRetriedTimes + 1);
                this.connectAndSubscribe();
                connFailRetriedTimes += 1;
            } else {
                this.stop();
                System.exit(0);
            }
        } else {
            connFailRetriedTimes = 0;
            log.info("海尔websocket认证成功");
        }
    }

    public void updateSubScribeResult(boolean subscribeResult) {
        if (!subscribeResult) {
            this.stop();
            log.error("海尔websocket订阅失败");
            System.exit(0);
        } else {
            log.info("海尔websocket订阅成功");
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        if (!haierWsProperties.isEnable()) {
            log.info("==================已配置【不启用】优家设备消息websocket客户端==================");
        } else {
            log.info("==================已配置【启用】优家设备消息websocket客户端==================");
            this.init();
        }
    }

    public void init() {
        boolean success = this.connectAndSubscribe();
        if (!success) {
            throw new RuntimeException("海尔websocket客户端启动失败");
        }
    }

    @OnMessage
    public void onMessage(final String message) {
        haierWsClientHeartBeatService.updateLastReceiveTime();
        haierWsMsgDispatcher.doDispatch(message, this);
    }

    @OnClose
    public void onClose(Session session, CloseReason closeReason) {
        log.info("海尔websocket Connection closed!");
        this.stop();
        // 连接异常关闭 CLOSED_ABNORMALLY(1006)，重启
        //  4500：通道异常，服务器端主动断开连接
        if (CloseReason.CloseCodes.CLOSED_ABNORMALLY.getCode() == closeReason.getCloseCode().getCode()
                || closeReason.getCloseCode().getCode() == 4500) {
            boolean reconnectSuccess = this.reconnectOnAbnormallyClose();
            if (!reconnectSuccess) {
                log.error("海尔websocket连接异常关闭，重连失败。");
            }
        } else {
            log.error("海尔websocket连接关闭:{}", closeReason.getCloseCode());
            dingTalkService.sendMessage(new TextContent("海尔websocket连接关闭:" + closeReason.getCloseCode()));
        }
    }

    /**
     * 处理异常关闭
     *
     * @return
     */
    private boolean reconnectOnAbnormallyClose() {
        Thread.interrupted();
        int delayTimeMultiplier = haierWsProperties.getReconnect().getMultiplier();
        int maxRetryTimes = haierWsProperties.getReconnect().getMaxRetryTimes();
        long delayTimeInSeconds = haierWsProperties.getReconnect().getDelayTime();
        for (int retriedTimes = 0; retriedTimes < maxRetryTimes; retriedTimes++) {
            try {
                // 第一次立即重试
                if (retriedTimes > 1) {
                    long waitSeconds = delayTimeInSeconds * delayTimeMultiplier;
                    log.info("海尔websocket尝试重连，等待：{}s...", waitSeconds);
                    TimeUnit.SECONDS.sleep(waitSeconds);
                    log.info("海尔websocket即将尝试重连...");
                }
                init();
                log.info("海尔websocket尝试重连，成功");
                return true;
            } catch (Exception e) {
                log.error("海尔websocket尝试重连，失败:", e);
                delayTimeMultiplier *= delayTimeMultiplier;
            }
        }
        log.error("海尔websocket尝试重连，失败:重试次数耗尽");
        return false;
    }

    /**
     * 发送控制信息
     *
     * @param command
     */
    public boolean sendCommand(String command) {
        if (!session.isOpen()) {
            log.error("海尔websocket控制消息发送异常:session closed");
            return false;
        }
        try {
            session.getBasicRemote().sendObject(command);
        } catch (Exception e) {
            log.error("海尔websocket控制消息发送异常:{}", e.getMessage());
            return false;
        }
        log.info("海尔websocket控制消息发送成功:{}", command);
        return true;
    }

    /**
     * 连接并订阅
     *
     * @return
     */
    private boolean connectAndSubscribe() {
        boolean connectSuccess = this.connect();
        if (!connectSuccess) {
            return false;
        }
        // 订阅
        return subscribe();
    }

    /**
     * 订阅
     *
     * @return
     */
    private boolean subscribe() {
        HaierBaseReqDTO<HaierWsSubReqDTO> subDataDTO = this.getSubDataDTO();
        String subscribeData = JSONObject.toJSONString(subDataDTO);
        log.info("发送订阅信息：{}", subscribeData);
        return this.sendCommand(subscribeData);
    }

    /**
     * 连接
     *
     * @return
     */
    private boolean connect() {
        String url = this.getServerConnectUrl();
        log.info("正在连接到海尔websocket服务器：{} ...", url);
        try {
            if (null == container) {
                container = ContainerProvider.getWebSocketContainer();
            }
            session = container.connectToServer(this, URI.create(url));
            // 值为0表示会话连接不会因为长时间无数据交互而超时
            session.setMaxIdleTimeout(0);
        } catch (Exception e) {
            log.error("海尔websocket连接失败：", e);
            return false;
        }
        return session.isOpen();
    }

    @Override
    public void destroy() throws Exception {
        this.stop();
    }

    /**
     * 停止
     */
    @SneakyThrows
    public void stop() {
        // 关闭 session 连接
        if (session.isOpen()) {
            session.close(new CloseReason(CloseReason.CloseCodes.NORMAL_CLOSURE, ""));
        }
    }

    /**
     * 获取ws服务端连接url
     *
     * @return
     */
    private String getServerConnectUrl() {
        StringBuilder wsServerUrlBuilder = new StringBuilder(haierWsProperties.getSchema());
        long timestamp = System.currentTimeMillis();
        String sign = UhomeUtils.getSign(haierWsProperties.getSystemId(), haierWsProperties.getSystemKey(), String.valueOf(timestamp));
        wsServerUrlBuilder
                .append(haierWsProperties.getHost()).append(haierWsProperties.getUri())
                .append("?")
                .append("systemId=").append(haierWsProperties.getSystemId())
                .append("&timestamp=").append(timestamp)
                .append("&sign=").append(sign)
                .append("&isSubscribeEarliestMessage=").append(haierWsProperties.getIsSubscribeEarliestMessage());
        return wsServerUrlBuilder.toString();
    }

    /**
     * 获取数据订阅DTO
     *
     * @return
     */
    private HaierBaseReqDTO<HaierWsSubReqDTO> getSubDataDTO() {
        List<HaierWsSubReqDTO.SubscribeData> subscribeDataList = haierWsProperties.getSubscribes().stream()
                .map(subscribe -> new HaierWsSubReqDTO.SubscribeData(subscribe.getTopic(), subscribe.getTypeIds()))
                .collect(Collectors.toList());
        return new HaierBaseReqDTO<>("subscribe", new HaierWsSubReqDTO(subscribeDataList));
    }

}
